package mapReduce.demo04_IOFormat.outputFormat;

import mapReduce.util.JobSubmit;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.UUID;

public class TextInMySQLOutApp {

    public static class TIMOMapper extends Mapper<LongWritable, Text,Text, IntWritable> {

        Text outKey = new Text();
        IntWritable outValue = new IntWritable();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {

            String line = value.toString();
            String[] fields = line.split(",");

            outKey.set(fields[0]);
            outValue.set(Integer.parseInt(fields[1]));

            context.write(outKey,outValue);

        }
    }

    public static void main(String[] args) {
        JobSubmit.submitBaseJob(TextInMySQLOutApp.class, args,MySQLOutputFormat.class);
    }
}

/**
 * 自定义输出格式化类，把结果数据写出到Mysql的表中
 */
class MySQLOutputFormat extends OutputFormat{

    @Override
    public RecordWriter<Text,IntWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new RecordWriter<Text,IntWritable>() {

            Connection conn = null;
            PreparedStatement ps = null;

            {
                try {
                    Class.forName("com.mysql.cj.jdbc.Driver");
                    conn = DriverManager.getConnection(
                            "jdbc:mysql://localhost:3308/bigdata2025",
                            "root",
                            "root"
                    );
                    ps = conn.prepareStatement("insert into student values (?,?,?)");
                }catch (Exception e){
                    e.printStackTrace();
                }
            }

            @Override
            public void write(Text key, IntWritable value) throws IOException, InterruptedException {
                try {
                    ps.setString(1, UUID.randomUUID().toString().toUpperCase());
                    ps.setString(2,key.toString());
                    ps.setInt(3,value.get());

                    int result = ps.executeUpdate();
                    if(result==1){
                        System.out.println("成功添加一条数据");
                    }else{
                        System.out.println("此条数据添加失败");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void close(TaskAttemptContext context) throws IOException, InterruptedException {
                try {
                    if(ps!=null){
                        ps.close();
                    }
                    if(conn!=null){
                        conn.close();
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }

            }
        };
    }

    @Override
    public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {

    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
        return new FileOutputCommitter(null,context);
    }
}
